package com.diver.flinkdemo.utils;

import com.alibaba.fastjson.JSONObject;
import com.diver.flinkdemo.entity.ResultConfig;
import com.diver.flinkdemo.entity.xxt.Message;
import com.diver.flinkdemo.entity.xxt.User;
import com.diver.flinkdemo.service.IHttpService;
import com.diver.flinkdemo.service.impl.HttpServiceImpl;
import com.diver.flinkdemo.service.impl.MessageService;
import lombok.extern.slf4j.Slf4j;
import org.apache.http.entity.ContentType;

import java.util.List;
import java.util.Map;
import java.util.concurrent.*;

/**
 * @Author liuC
 * @Date 2023/4/17 0017 15:01
 * @Describe
 */
@Slf4j
public class FlinkThreadUtil {
    /*static int a=0;
    static int count=2000;
    public static void main(String[] args) throws InterruptedException {
        ExecutorService service = Executors.newCachedThreadPool();
        //闭锁 在一些条件下可放开  参数:加多少把锁
        CountDownLatch countDownLatch=new CountDownLatch(count);
        //信号量
        Semaphore semaphore=new Semaphore(1);
        for(int i=0;i<count;i++){
            service.execute(new Runnable() {
                @Override
                public void run() {
                    try {  //拿走一个信号
                        semaphore.acquire();
                        a++;
                        //解一把锁
                        countDownLatch.countDown();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }finally {
                        //释放信号
                        semaphore.release();
                    }
                }
            });
        }
        service.shutdown();
        //会进入阻塞状态  什么时候把锁全解了   阻塞状态才会解除
        countDownLatch.await();
        System.out.println(a);
        //2000
    }*/

    public ThreadPoolExecutor threadPoolExecutor () {
        //核心线程数
        int corePoolSize = 8;
        //最大线程数
        int maximumPoolSize = 16;
        //线程无引用存活时间
        long keepAliveTime = 60000;
        //时间单位,毫秒
        TimeUnit unit = TimeUnit.MILLISECONDS;
        //任务队列
        BlockingQueue<Runnable> runnableBlockingQueue = new ArrayBlockingQueue<>(1000);
        ThreadFactory threadFactory = Executors.defaultThreadFactory();
        //拒绝执行处理器
        RejectedExecutionHandler executionHandler = new ThreadPoolExecutor.CallerRunsPolicy();
        ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(corePoolSize, maximumPoolSize, keepAliveTime, unit, runnableBlockingQueue, threadFactory, executionHandler);
        return threadPoolExecutor;
    }

    /*public void sendTrdRequest (String url, Map<String, String> headers, String body,
                                                ContentType contentType, String charset) {
        threadPoolExecutor().submit(new Runnable() {
            @Override
            public void run() {
                ResultConfig<String> result = httpService.doStringEntityPost(url, headers, body, contentType, charset);
                log.info("第三方接口调用{} 返回:{}", url, result);
            }
        });
    }*/
}
